package org.example.flink.source;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import scala.Tuple2;
import scala.Tuple3;

/**
 * @author zhangjungang
 * @desc 时间窗口测试
 * @date 2021/10/12 上午10:38
 */
public class S3 implements SourceFunction<Tuple3<String, Date,Integer>> {

    static final String[] NAMES = {"tom", "jerry", "alice", "bob", "john", "grace"};
    private static final Random rnd = new Random();

    @Override
    public void run(SourceContext<Tuple3<String, Date,Integer>> sourceContext) throws Exception {

        Map<String,Boolean> map = new HashMap<>();

        int num = 20;
        for (int i = 0; i < num; i++) {
            String name = NAMES[rnd.nextInt(6)];
            Boolean flag = map.get(name);
            Tuple3 tuple3 = null;
            if(flag == null || flag == true){
                tuple3 = Tuple3.apply(name,new Date(),0);
                map.put(name,false);
            }else {
                tuple3 = Tuple3.apply(name,new Date(),1);
                map.put(name,true);
            }
            sourceContext.collect(tuple3);
            TimeUnit.MILLISECONDS.sleep(500);
        }
        System.out.println("s3-------end");
    }

    @Override
    public void cancel() {

    }
}
